package com.example.demo.MyBlockingQueue;

import java.util.Arrays;
import java.util.concurrent.TimeUnit;
import java.util.concurrent.atomic.AtomicInteger;
import java.util.concurrent.locks.Condition;
import java.util.concurrent.locks.ReentrantLock;

/**
 * Created with IntelliJ IDEA.
 * Description:
 * User: sfx
 * Date: 2023-09-11
 * Time: 17:19
 */
public class MyBlockingQueue3<E> implements BlockingQueue<E> {
    // 队列主体
    private final E[] array;
    // 头指针
    private int head;
    // 尾指针
    private int tail;
    // 实际元素个数
    // 因为现在是两把锁,为了保证线程安全,保证size的原子性,所以使用原子类
    private AtomicInteger size = new AtomicInteger(0);
    // 锁对象->配合offer使用
    private ReentrantLock tailLock = new ReentrantLock();
    // 锁对象->配合poll使用
    private ReentrantLock headLock = new ReentrantLock();
    // 头waits ->配合poll使用,当为空了进入headWaits
    private Condition headWaits = headLock.newCondition();
    // 尾waits -->配合offer使用,当慢了,进入tailWaits
    private Condition tailWaits = tailLock.newCondition();

    public MyBlockingQueue3(int capacity) {
        array = (E[]) new Object[capacity];
    }

    private boolean isFull() {
        return size.get() == array.length;
    }

    private boolean isEmpty() {
        return size.get() == 0;
    }

    @Override
    public void offer(E e) throws InterruptedException {
        tailLock.lockInterruptibly();
        int c;
        try {
            while (isFull()) {
                tailWaits.await();//offer1,offer2,offer3
            }
            array[tail] = e;
            if (++tail == array.length) {
                tail = 0;
            }
            c = size.getAndIncrement();
            // 进行级联唤醒
            if(c + 1 < array.length) {
                //还没有满,自己唤醒自己
                tailWaits.signal();
            }

        } finally {
            tailLock.unlock();
        }
        // 唤醒等待poll的线程,防止死锁,所以放在外边,让两把锁平级
        // offer1,offer2,offer3
        if(c == 0) {
            headLock.lock();
            try {
                headWaits.signal();
            } finally {
                headLock.unlock();
            }
        }
    }

    @Override
    public boolean offer(E e, long timeout) throws InterruptedException {
        tailLock.lockInterruptibly();
        long nanos = TimeUnit.MILLISECONDS.toNanos(timeout);
        try {
            while (isFull()) {
                if (nanos <= 0) {
                    return false;
                }
                nanos = tailWaits.awaitNanos(nanos);
            }
            array[tail] = e;
            if (++tail == array.length) {
                tail = 0;
            }
            size.getAndIncrement();
        } finally {
            tailLock.unlock();
        }
        headLock.lock();
        try{
            headWaits.signal();
        }finally {
            headLock.unlock();
        }
        return true;
    }

    @Override
    public E poll() throws InterruptedException {
        headLock.lockInterruptibly();
        E e;
        int c;
        try {
            while (isEmpty()) {
                headWaits.await();//poll_1,poll_2,poll_3
            }
            // 取出对头元素
            e = array[head];
            array[head] = null;//方便GC
            if (++head == array.length) {
                head = 0;
            }
            c = size.getAndDecrement();
            if(c + 1 > 0) {
                // 还有元素,自己唤醒自己
                headWaits.signal();
            }
        } finally {
            headLock.unlock();
        }
        // 减少唤醒时加锁的次数
        // poll_1,poll_1,poll_3
        // 唤醒正在等待的offer线程,防止死锁放在外边,与它headLock进行评级
        // 从满->不满的时候进行唤醒
        if(c == array.length) {
            tailLock.lock();
            try {
                tailWaits.signal();
            } finally {
                tailLock.unlock();
            }
        }
        return e;
    }

    @Override
    public String toString() {
        return "MyBlockingQueue{" +
                "queue=" + Arrays.toString(array);
    }
}